package State;

import bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Spring_Hu
 * @date 2021/10/15 19:06
 */
public class Operator_state_Broadcast {
    public static void main(String[] args) throws Exception {
        //算子状态的定义 broadcast 实现流的相互转换
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        DataStreamSource<String> mainSource = env
                .socketTextStream("localhost", 9999);
        DataStreamSource<String> broadcastSource = env
                .socketTextStream("localhost", 8888);


        //将要广播的数据输入到广播流中
        MapStateDescriptor<String, String> a = new MapStateDescriptor<>("broadcastState", Types.STRING, Types.STRING);
        BroadcastStream<String> broadcastStream = broadcastSource.broadcast(a);

        //将两个流连接

        mainSource.connect(broadcastStream).process(new BroadcastProcessFunction<String, String, String>() {
            @Override
            public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
               //从上下文中取出 广播流的数据
                String s = ctx.getBroadcastState(a).get("a");
                if (s!=null) {
                    if(s.equals("0")){
                        out.collect("切换到0号线路。。。。。");
                    }else if(s.equals("1")){
                        out.collect("切换到1号线路。。。。。");
                    }else{
                        out.collect("线路错误！");
                    }
                }
            }

            @Override
            public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
                ctx.getBroadcastState(a).put("a",value);
            }
        }).print();

        env.execute();
    }
    private static class MyProcess extends ProcessFunction<WaterSensor, String> implements CheckpointedFunction{
        String statevalue;
        ListState<String> listState;

        @Override
        //Checkpoint时会调用这个方法，我们要实现具体的snapshot逻辑，比如将哪些本地状态持久化
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("snapshotState......");

            listState.add(statevalue);

        }



        //初始化状态 向本地状态中填充数据. 每个子任务调用一次
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("initializeState......");
            listState = context.getOperatorStateStore().getListState(new ListStateDescriptor("list_state", Types.STRING));
            for (String s : listState.get()) {
                System.out.println("拿到的状态值："+s);
            }
        }

        @Override
        public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
            System.out.println("processElement....");
            statevalue=value.toString();
            out.collect(value.toString());
        }
    }



}
